package com.lzq.mq.broker.event.spi.listener;

import com.alibaba.fastjson2.JSON;
import com.lzq.mq.broker.cache.CommonCache;
import com.lzq.mq.broker.event.model.ConsumeMsgAckEvent;
import com.lzq.mq.broker.model.MqTopicModel;
import com.lzq.mq.broker.rebalance.ConsumerInstance;
import com.lzq.mq.common.coder.TcpMsg;
import com.lzq.mq.common.dto.ConsumeMsgAckReqDTO;
import com.lzq.mq.common.dto.ConsumeMsgAckRespDTO;
import com.lzq.mq.common.enums.AckStatus;
import com.lzq.mq.common.enums.BrokerResponseCode;
import com.lzq.mq.common.event.Listener;
import com.lzq.mq.common.utils.CollectionUtils;
import lombok.extern.slf4j.Slf4j;

import java.util.List;
import java.util.Map;

@Slf4j
public class ConsumeMsgAckListener implements Listener<ConsumeMsgAckEvent> {

    @Override
    public void onReceive(ConsumeMsgAckEvent event) throws Exception {
        ConsumeMsgAckReqDTO consumeMsgAckReqDTO = event.getConsumeMsgAckReqDTO();
        String topic = consumeMsgAckReqDTO.getTopic();
        String consumeGroup = consumeMsgAckReqDTO.getConsumeGroup();
        Integer queueId = consumeMsgAckReqDTO.getQueueId();
        Integer ackCount = consumeMsgAckReqDTO.getAckCount();

        // 构建ack响应
        ConsumeMsgAckRespDTO consumeMsgAckRespDTO = new ConsumeMsgAckRespDTO();
        consumeMsgAckRespDTO.setMsgId(event.getMsgId());

        // 处理ack
        MqTopicModel mqTopicModel = CommonCache.getMqTopicModelMap().get(topic);
        if (mqTopicModel == null) {
            //topic不存在，ack失败
            consumeMsgAckRespDTO.setAckStatus(AckStatus.FAIL.getCode());
            event.getChannelHandlerContext().writeAndFlush(new TcpMsg(BrokerResponseCode.BROKER_UPDATE_CONSUME_OFFSET_RESP.getCode(),
                    JSON.toJSONBytes(consumeMsgAckRespDTO)));
            return;
        }
        // 获取 每个topic的各个消费者组的各个消费者
        Map<String, List<ConsumerInstance>> consumerInstanceMap = CommonCache.getConsumeHoldMap().get(topic);
        if (consumerInstanceMap == null || consumerInstanceMap.isEmpty()) {
            // topic不存在，ack失败
            consumeMsgAckRespDTO.setAckStatus(AckStatus.FAIL.getCode());
            event.getChannelHandlerContext().writeAndFlush(new TcpMsg(BrokerResponseCode.BROKER_UPDATE_CONSUME_OFFSET_RESP.getCode(),
                    JSON.toJSONBytes(consumeMsgAckRespDTO)));
            return;
        }
        // 获取当前消费者组的实体
        List<ConsumerInstance> consumeGroupInstances = consumerInstanceMap.get(consumeGroup);
        if (CollectionUtils.isEmpty(consumeGroupInstances)) {
            // topic不存在当前消费者组, ack失败
            consumeMsgAckRespDTO.setAckStatus(AckStatus.FAIL.getCode());
            event.getChannelHandlerContext().writeAndFlush(new TcpMsg(BrokerResponseCode.BROKER_UPDATE_CONSUME_OFFSET_RESP.getCode(),
                    JSON.toJSONBytes(consumeMsgAckRespDTO)));
            return;
        }
        // 获取当前消费者组下的消费者实例
        String currentConsumeReqId = consumeMsgAckReqDTO.getIp() + ":" + consumeMsgAckReqDTO.getPort();
        ConsumerInstance matchInstance = consumeGroupInstances.stream()
                .filter(item -> item.getConsumerReqId().equals(currentConsumeReqId)).findAny().orElse(null);
        if (matchInstance == null) {
            // 当前消费者实例不存在，ack失败(主要是校验消费者实例)
            /*
            在发生重平衡之后可能发生此种错误, 即当前节点已经被剔除
             */
            consumeMsgAckRespDTO.setAckStatus(AckStatus.FAIL.getCode());
            event.getChannelHandlerContext().writeAndFlush(new TcpMsg(BrokerResponseCode.BROKER_UPDATE_CONSUME_OFFSET_RESP.getCode(),
                    JSON.toJSONBytes(consumeMsgAckRespDTO)));
            return;
        }
        // 数据的ack，到底应该客户端传递offset过来好 还是在服务端计算offset值好？(批量ack)
        for (int i = 0; i < ackCount; i++) {
            CommonCache.getConsumeQueueConsumeHandler().ack(topic, consumeGroup, queueId);
        }
        log.info("收到ack ,topic - {},consumeGroup - {},queueId - {},ackCount - {}",
                topic, consumeGroup, queueId, ackCount);
        // 回复ack成功
        consumeMsgAckRespDTO.setAckStatus(AckStatus.SUCCESS.getCode());
        TcpMsg tcpMsg = new TcpMsg(BrokerResponseCode.BROKER_UPDATE_CONSUME_OFFSET_RESP.getCode(),
                JSON.toJSONBytes(consumeMsgAckRespDTO));
        event.getChannelHandlerContext().writeAndFlush(tcpMsg);
    }
}

